package kassiopeia.plugin.crawler.store;

import java.net.URL;
import java.util.List;

import kassiopeia.plugin.Event;
import kassiopeia.plugin.PluginException;
import kassiopeia.plugin.ProcessorPlugin;

import org.apache.log4j.Logger;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.hibernate.Transaction;
import org.hibernate.cfg.Configuration;
import org.hibernate.tool.hbm2ddl.SchemaUpdate;

public class Store extends ProcessorPlugin {

    private static final Logger logger = Logger.getLogger(Store.class);

    // Outputs
    public static final String URL_OUTPUT = "out";

    // Event headers
    public static final String URL_HEADER_NAME = "URL";
    public static final String MIME_TYPE_HEADER_NAME = "MIME_TYPE";

    // Parameters
    public static final String CONNECTION_URL_PARAM = "connection-url";
    public static final String DRIVER_CLASS_PARAM = "driver-class";
    public static final String DIALECT_NAME_PARAM = "dialect-name";

    private SessionFactory sessionFactory;
    private String connectionUrl;
    private String driverClass;
    private String dialectName;

    @Override
    public void beforeTask() {
        connectionUrl = getContext().getParameter(CONNECTION_URL_PARAM);
        driverClass = getContext().getParameter(DRIVER_CLASS_PARAM);
        dialectName = getContext().getParameter(DIALECT_NAME_PARAM);

        Configuration configuration = createHibernateConfiguration();
        initializeSessionFactory(configuration);
        updateDatabaseSchema(configuration);
    }

    @Override
    public void process(Event event) {
        CrawledResource crawledResource = getCrawledResource(event);
        Session session = sessionFactory.openSession();

        Transaction transaction = null;
        try {
            transaction = session.beginTransaction();
            session.save(getCrawledResource(event));
            transaction.commit();

            logger.info("Resource stored [" + crawledResource + "]");

        } catch(Exception exception) {
            if(transaction != null) {
                transaction.rollback();
            }
            throw new PluginException("Can't store event.",exception);
        } finally {
            session.close();
        }
    }

    @Override
    public void afterTask() {
        try {
            if(sessionFactory != null) {
                sessionFactory.close();
            }
            logger.info("Database connection closed.");
        } catch (Exception exception) {
            throw new PluginException("Can't create Hibernate Session.",exception);
        }
    }

    private void initializeSessionFactory(Configuration configuration) {
        sessionFactory = configuration.buildSessionFactory();

        logger.info("Connected to database [" + connectionUrl + "]");
        logger.info("Hibernate dialect: " + dialectName);
        logger.info("Database driver: " + driverClass);
    }

    private Configuration createHibernateConfiguration() {
        Configuration configuration = new Configuration()
            .addAnnotatedClass(CrawledResource.class)
            .setProperty("hibernate.dialect", dialectName)
            .setProperty("hibernate.connection.driver_class", driverClass)
            .setProperty("hibernate.connection.url", connectionUrl)
            .setProperty("hibernate.c3p0.min_size", "5")

            .setProperty("hibernate.show_sql", "false")
            .setProperty("hibernate.format_sql", "false")
            .setProperty("hibernate.c3p0.max_size", "20")
            .setProperty("hibernate.c3p0.timeout", "1800")
            .setProperty("hibernate.c3p0.max_statements", "50")
            .setProperty("hibernate.order_updates", "true");
        return configuration;
    }

    @SuppressWarnings("unchecked")
    private void updateDatabaseSchema(Configuration configuration) {
        logger.info("Updating database schema...");

        SchemaUpdate schemaUpdate = new SchemaUpdate(configuration);
        schemaUpdate.execute(true, true);

        List<Exception> exceptions = schemaUpdate.getExceptions();
        if(exceptions.size() > 0) {
            throw new PluginException("Can't update database schema.",exceptions.get(0));
        }

        logger.info("Database schema updated.");
    }

    private CrawledResource getCrawledResource(Event event) {
        byte[] pageContent = getNotNullableEventBody(event, byte[].class);
        URL url = getNotNullableEventHeader(event, URL_HEADER_NAME, URL.class);
        String urlString = url.toExternalForm();
        String contentMimeType = event.getHeader(MIME_TYPE_HEADER_NAME, String.class);

        CrawledResource crawledResource = new CrawledResource();
        crawledResource.setContent(pageContent);
        crawledResource.setUrl(urlString);
        crawledResource.setSize(pageContent.length);
        crawledResource.setMimeType(contentMimeType);

        return crawledResource;
    }
}
